package com.huayi.breakpoint_continuation_interface.controller;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.huayi.breakpoint_continuation_interface.bean.BreakpointStats;
import com.huayi.breakpoint_continuation_interface.bean.User;
import com.huayi.breakpoint_continuation_interface.service.Breakpoint_Continuation_FromCk_Service;
import com.huayi.breakpoint_continuation_interface.utils.JSONUtil;
import com.huayi.breakpoint_continuation_interface.utils.RedisUtil;
import com.huayi.breakpoint_continuation_interface.utils.TokenUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.*;
import redis.clients.jedis.Jedis;

import javax.servlet.http.Cookie;
import javax.servlet.http.HttpServletResponse;
import javax.servlet.http.HttpSession;
import java.lang.reflect.Field;
import java.util.List;
import java.util.Set;

/**
 * @Description :供外部从kafka发送数据接口以及从ClickHouse中获取结果数据
 * @Author:yujiahao
 * @Date: 2022/5/31 9:33
 * @Version: 1.0
 */

@RestController
public class SuperController {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    @Autowired
    private Breakpoint_Continuation_FromCk_Service breakpointContinuationFromCkService;

    //获取Redis连接
    Jedis jedis = RedisUtil.getJedis();


    /**
     * TODO 1、用于获取token并存入redis
     *
     * @param user
     * @param session
     * @param response
     * @return
     */
    @GetMapping("/login")
    public Object login(User user, HttpSession session, HttpServletResponse response) {

        //从mysql中验证用户名密码是否匹配
        User loginUser = breakpointContinuationFromCkService.getAccount(user);
        JSONObject jsonObject = new JSONObject();
        if (loginUser != null && user.getUser_name().length() > 0 && user.getPassword().length() > 0) {
            session.setAttribute("loginUser", loginUser);
            String token = breakpointContinuationFromCkService.getToken(loginUser);
            jsonObject.put("token", token);
            Cookie cookie = new Cookie("token", token);
            cookie.setPath("/");
            response.addCookie(cookie);
            //将token存入Redis中缓存
            String token_key = token.substring(0, 10);
            jedis.set(token_key, token);
            //Redis中设置token过期时间为10分钟
            jedis.expire(token_key, 600);
            return jsonObject;
        } else {
            jsonObject.put("message", "登录失败,密码错误");
        }
        return jsonObject;
    }


    /**
     * 视频断点续传从kafka接收数据接口,需要获取有效token才可
     *
     * @param topic
     * @param data
     * @return
     */

    @PostMapping("/huayi/breakpointContinuation")
    public String breakpointContinuation(@RequestParam(value = "topic", defaultValue = "breakpotint_ck") String topic, @RequestBody String data) {
        //从header中解析token并拼接key
        String token = TokenUtil.getRequest().getHeader("token");
        if (token.isEmpty()) {
            return "token为空，请先获取！";
        }
        String subToken = token.substring(0, 10);
        //在从redis中查询token 并验证是否正确或有效
        String getValue = jedis.get(subToken);
        if (token.equals(getValue) && token.length() > 0 && getValue.length() > 0) {

            String str = data.toString();
            //判断是否则JSON串，否则返回失败
            if (JSONUtil.isJSON(str) == false) {
                return "failed!";
            }
            JSONObject jsonObject = JSON.parseObject(str);

            if (jsonObject.getString("id").isEmpty() || jsonObject.getString("id").length() == 0 ||
                    jsonObject.getString("platform_id").isEmpty() || jsonObject.getString("platform_id").length() == 0 ||
                    jsonObject.getString("user_id").isEmpty() || jsonObject.getString("user_id").length() == 0 ||
                    jsonObject.getString("course_id").isEmpty() || jsonObject.getString("course_id").length() == 0 ||
                    jsonObject.getString("play_current").isEmpty() || jsonObject.getString("play_current").length() == 0 ||
                    jsonObject.getString("play_insert").isEmpty() || jsonObject.getString("play_insert").length() == 0 ||
                    jsonObject.getString("play_total").isEmpty() || jsonObject.getString("play_total").length() == 0 ||
                    jsonObject.getString("identify_id").isEmpty() || jsonObject.getString("identify_id").length() == 0

            ) {
                return "failed!";
            }

            //todo 1.1、发送数据到broker主题
            ListenableFuture<SendResult<String, String>> send = kafkaTemplate.send(topic, str);
            //todo 1.2、带回调，异步等待结果
            send.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                @Override
                public void onFailure(Throwable ex) {
                    System.out.println("消息发送失败：" + ex.getMessage().toString());
                }

                @Override
                public void onSuccess(SendResult<String, String> result) {
                    System.out.println("消息成功发送到主题：" + result.getRecordMetadata().topic() + " 中");

                }
            });
            return "success!";
        }
        return "token过期，请重新获取！";

    }


    //TODO 视频断点续传 ==> 从Click House中获取结果数据

    @GetMapping("/huayi/breakpointContinuationFromCk")
    public String getAllFromCk() {
        List<BreakpointStats> list = breakpointContinuationFromCkService.getAll();
        StringBuilder stringBuilder = new StringBuilder();
        for (int i = 0; i < list.size(); i++) {
            BreakpointStats breakpointTestStats = list.get(i);
            //如结果为多条数据中间可使用逗号分割
            if (i >= 1) {
                stringBuilder.append(",");
            }
            if (breakpointTestStats.getUser_id() == null || breakpointTestStats.getPlatform_id() == null ||
                    breakpointTestStats.getCourse_id() == null || breakpointTestStats.getPlay_current() == null ||
                    breakpointTestStats.getPlay_insert() == null) {
                return "failed!";
            }
            //最终返回符合要求的结果为JSON串
            stringBuilder.append("{" +
                    "\"id\":\"" +
                    breakpointTestStats.getId() +
                    "\",\"platform_id\":\"" +
                    breakpointTestStats.getPlatform_id() +
                    "\",\"user_id\":\"" +
                    breakpointTestStats.getUser_id() +
                    "\",\"course_id\":\"" +
                    breakpointTestStats.getCourse_id() +
                    "\",\"play_current\":\"" +
                    breakpointTestStats.getPlay_current() +
                    "\",\"play_insert\":\"" +
                    breakpointTestStats.getPlay_insert() +
                    "\",\"play_total\":\"" +
                    breakpointTestStats.getPlay_total() +
                    "\",\"identify_id\":\"" +
                    breakpointTestStats.getIdentify_id() +
                    "\"}"
            );

        }
        String str = stringBuilder.toString();
        //判断是否则JSON串，否则返回失败,正式环境会有一条数据，测试逻辑是全表扫描所以会报错无法解析
//        if (JSONUtil.isJSON(str) == false) {
//            System.out.println(str);
//            return "failed!";
//        }
        return str;
    }
}
